package tk.xboot.flink.kfk.json2;

import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.watermark.Watermark;

import javax.annotation.Nullable;
import java.text.SimpleDateFormat;


public class MessageWaterEmitter implements AssignerWithPeriodicWatermarks<ObjectNode> {


    private long maxOutOfOrderness = 1000;

    private long currentMaxTimestamp=0L;

    private Watermark watermark;
    SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");


    /**
     * Returns the current watermark. This method is periodically called by the
     * system to retrieve the current watermark. The method may return {@code null} to
     * indicate that no new Watermark is available.
     *
     * <p>The returned watermark will be emitted only if it is non-null and its timestamp
     * is larger than that of the previously emitted watermark (to preserve the contract of
     * ascending watermarks). If the current watermark is still
     * identical to the previous one, no progress in event time has happened since
     * the previous call to this method. If a null value is returned, or the timestamp
     * of the returned watermark is smaller than that of the last emitted one, then no
     * new watermark will be generated.
     */
    @Nullable
    @Override
    public Watermark getCurrentWatermark() {
        watermark = new Watermark(currentMaxTimestamp - maxOutOfOrderness);
        return watermark;
    }

    /**
     * Assigns a timestamp to an element, in milliseconds since the Epoch.
     *
     * <p>The method is passed the previously assigned timestamp of the element.
     * That previous timestamp may have been assigned from a previous assigner,
     * by ingestion time. If the element did not carry a timestamp before, this value is
     * {@code Long.MIN_VALUE}.
     *
     * @param element                  The element that the timestamp will be assigned to.
     * @param previousElementTimestamp The previous internal timestamp of the element,
     *                                 or a negative value, if no timestamp has been assigned yet.
     * @return The new timestamp.
     */
    @Override
    public long extractTimestamp(ObjectNode element, long previousElementTimestamp) {

        if(element != null && element.get("ts") != null){
            long ts = element.get("ts").longValue();

//            if (watermark != null) {
//                System.out.println("timestamp : " + ts + "|" + format.format(ts)
//                        + " currentMaxTimestamp : " + currentMaxTimestamp + "|"
//                        + format.format(currentMaxTimestamp) + "," + " watermark : " + watermark.getTimestamp() + "|"
//                        + format.format(watermark.getTimestamp()));
//            }
            currentMaxTimestamp =  Math.max(ts, currentMaxTimestamp);
            return ts;
        }
        else
            throw new IllegalArgumentException("element's ts field must not be null");
    }
}
